package com.dingwen.treasure.canal;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * 简单模式，canal demo: TODO 业务模拟、完善
 *
 * @author dingwen
 */
@SuppressWarnings("ALL")
public class SimpleCanalClientExample {
    public static void main(String[] args) throws Exception {
        // 1. 获取连接
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111),
                "example",
                "",
                ""
        );
        while (Boolean.TRUE) {
            // 2. 连接
            canalConnector.connect();
            // 3. 订阅数据库
            canalConnector.subscribe(".*\\..*");
            // 4. 获取数据
            // 批量获取最多一百条数据
            Message message = canalConnector.get(100);

            // 5. 获取Entry
            List<CanalEntry.Entry> entries = message.getEntries();


            if (entries.isEmpty() || entries.size() == 0) {
                Thread.sleep(1000);
                System.out.println("没有抓取到数据，休息一会儿～");
            } else {
                // 6. 遍历解析entry
                entries.forEach(entry -> {
                    //7. 获取表名称
                    String tableName = entry.getHeader().getTableName();
                    // 8. 获取类型
                    CanalEntry.EntryType entryType = entry.getEntryType();
                    // 9. 获取序列化之后的数据
                    ByteString storeValue = entry.getStoreValue();
                    // 10. 判断entry类型
                    if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                        // 11. 反序列化数据
                        try {
                            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

                            // 12. 获取数据集
                            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

                            // 13. 当前事件的类型
                            CanalEntry.EventType eventType = rowChange.getEventType();

                            JSONObject jsonObjectBefore = new JSONObject();
                            JSONObject jsonObjectAfter = new JSONObject();
                            rowDatasList.forEach(rowData -> {
                                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                                beforeColumnsList.forEach(column -> {
                                    jsonObjectBefore.put(column.getName(), column.getValue());
                                });
                                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                                afterColumnsList.forEach(column -> {
                                    jsonObjectAfter.put(column.getName(), column.getValue());
                                });
                            });
                            System.out.println("eventType = " + eventType);
                            System.out.println("jsonObjectBefore = " + jsonObjectBefore);
                            System.out.println("jsonObjectAfter = " + jsonObjectAfter);
                            System.out.println("tableName = " + tableName);
                            System.out.println("entryType = " + entryType);
                        } catch (InvalidProtocolBufferException e) {
                            e.printStackTrace();
                        }
                    } else {
                        System.out.println("entryType = " + entryType);
                    }
                });
            }
        }
    }
}
